/*
 * JobPoll.cpp
 *
 *   Created on: 2010-3-30     
 *  Description: 
 *    Copyright: 2010 @ ICT Li Heyuan
 */

#include <iostream>
using std::cout;
using std::endl;

#include "../../include/thread/JobPoll.h"
#include "../../include/util/JsonRpcException.h"
#include "../../include/server/TcpServer.h"

//static成员初始化
JobPoll* JobPoll::_pInstance = NULL;
unsigned int JobPoll::_numThreads = 0;
unsigned int JobPoll::_numAliveThreads = 0;
list<Job> JobPoll::_jobs;
pthread_mutex_t JobPoll::_mutex_jobs =
PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t JobPoll::_cond_new_job =
PTHREAD_COND_INITIALIZER;

JobPoll::JobPoll(unsigned int n)
{
	//默认Fast类型的mutex
	pthread_mutex_init(&_mutex_jobs, NULL);
	//创建默认的cond
	pthread_cond_init(&_cond_new_job, NULL);
	_numThreads = n;

	initThreads();
}

JobPoll::~JobPoll()
{
	pthread_cond_destroy(&_cond_new_job);
	pthread_mutex_destroy(&_mutex_jobs);
}

JobPoll * JobPoll::getInstance(unsigned int n /* = 2*/)
{
	if (_pInstance == NULL)
	{
		_pInstance = new JobPoll(n);
		if (_pInstance == NULL)
		{
			throw JsonRpcException("memory alloc(JobPoll) fail");
		}
	}

	return _pInstance;
}

void JobPoll::addJob(Conn *conn)
{
	pthread_mutex_lock(&_mutex_jobs);

	bool needSignal = _jobs.empty();
	_jobs.push_back(Job(conn));

	if (needSignal)
	{
		pthread_cond_signal(&_cond_new_job);
	}

	pthread_mutex_unlock(&_mutex_jobs);
}

void* JobPoll::thread_func_worker(void *arg)
{
	JobPoll *pJobPoll = (JobPoll*) arg;
	while (true)
	{
		Job job(NULL);
		//获取一个Job
		pthread_mutex_lock(&_mutex_jobs);
		while (pJobPoll->_jobs.empty())
		{
			//			printf("wait in thread %lu\n", pthread_self());
			pthread_cond_wait(&_cond_new_job, &_mutex_jobs);
		}
		job = pJobPoll->_jobs.front();
		pJobPoll->_jobs.pop_front();
		pthread_mutex_unlock(&_mutex_jobs);

		try
		{
			if (job._pConn == NULL)
			{
				continue;
			}

			job._pConn->_server->getNetLogger().info("execute in pid:%lu",
					pthread_self());

			//执行method
			job ._pConn->callMethod();
			job._pConn->_server->getNetLogger().info("execute finish :%lu",
					pthread_self());
			//执行完毕后，conn回调，然后模拟触发一次EPOLL_OUT
			if (!job._pConn->handle(FINISH))
			{
				//如果预定Server先关闭
				if (job._pConn->_server->isCloseFirst())
				{
					job._pConn->close();
				}
			}

			//全部执行完毕，可安全关闭了
			job._pConn->setSafeDel(true);
		} //end try
		catch (...)
		{
			//说明job._pConn已经被删除，不做任何操作
			cout << "in jobpool,has been deleted" << endl;
			//可以安全关闭
			job._pConn->setSafeDel(true);
		}
	}//end while
	return NULL;
}

void JobPoll::initThreads()
{
	//创建线程参数DETACHED，线程结束后自己销毁资源
	pthread_attr_t attr;
	pthread_attr_init(&attr);
	if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
	{
		//设置DETACHED属性失败
		throw JsonRpcException("set detached to pthread_attr_t fail");
	}
	/*	//获取调度类型
	 int type = 0;
	 if (pthread_attr_getschedpolicy(&attr, &type))
	 {
	 //获取schedpolicy
	 throw JsonRpcException("get schedpolicy fail");
	 }
	 //设置优先级(Bug)
	 struct sched_param param;
	 param.__sched_priority = (sched_get_priority_max(type)
	 + sched_get_priority_min(type)) / 2;
	 if (pthread_attr_setschedparam(&attr, &param))
	 {
	 //设置优先级失败属性失败
	 throw JsonRpcException("set sched_param to pthread_attr_t fail");
	 }*/

	for (unsigned int i = 0; i < _numThreads; i++)
	{
		pthread_t tid;
		if (pthread_create(&tid, &attr, thread_func_worker, NULL))
		{
			//创建线程
			throw JsonRpcException("pthread_create fail");
		}
		else
		{
			//创建线程成功
			_numAliveThreads++;
		}
	}
	//销毁线程参数
	pthread_attr_destroy(&attr);
}
